package com.gitee.sparta.core.process.operate.redis;

import com.gitee.sparta.common.SpartaConstant;
import com.gitee.sparta.common.SpartaLuaConstant;
import com.gitee.sparta.common.helpers.JacksonHelper;
import com.gitee.sparta.common.helpers.ScriptShaHelper;
import com.gitee.sparta.common.helpers.ZipStrHelper;
import com.gitee.sparta.core.bean.metadata.Metadata;
import com.gitee.sparta.core.process.feature.SpartaWorkerFeature;
import com.gitee.sparta.core.process.operate.ReadableTabOperate;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RScript;
import org.redisson.api.RedissonClient;
import org.redisson.client.codec.LongCodec;
import org.redisson.client.codec.StringCodec;

import java.util.*;

/**
 * 使用 lua 针对 readable_tab 的读取实现
 *
 * @param <V>
 * @author zyred
 * @since v 0.0.1
 */
@Slf4j
public class LuaReadableTabOperate<V> implements ReadableTabOperate<V> {

    private final RedissonClient redissonClient;
    private final SpartaCoreProperties properties;

    public LuaReadableTabOperate(SpartaCoreProperties properties, RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
        this.properties = properties;
    }

    @Override
    public void delReadableTab(String fullTopic, String readableTabKey) {
        Map<String, String> cache = SpartaLuaConstant.getSha(SpartaConstant.DEL_HAVE_READ_IN_READABLE_TAB);

        try {
            RScript script = this.redissonClient.getScript(new StringCodec());
            script.evalSha(RScript.Mode.READ_ONLY, cache.get(SpartaConstant.LUA_SHA_KEY),
                    RScript.ReturnType.VALUE, Collections.singletonList(readableTabKey), fullTopic);
        } catch (Exception ex) {
            log.error("[Sparta] Del have read failed to execute, topic: {}, exception: {}",
                    fullTopic, ex.getMessage(), ex);
            ScriptShaHelper.reloadScriptSha(ex, this.redissonClient, this.properties);
        }
    }

    @Override
    public void delMetadataReadableTab(String fullTopic, String readableTabKey, String metadataTabKey) {
        Map<String, String> cache = SpartaLuaConstant.getSha(SpartaConstant.DEL_METADATA_READABLE_TAB);

        final List<Object> keys = Arrays.asList(readableTabKey, metadataTabKey);

        try {
            RScript script = this.redissonClient.getScript(new StringCodec());
            script.evalSha(RScript.Mode.READ_ONLY, cache.get(SpartaConstant.LUA_SHA_KEY),
                    RScript.ReturnType.VALUE, keys, fullTopic);
        } catch (Exception ex) {
            log.error("[Sparta] Del metadata readable failed to execute, topic: {}, exception: {}",
                    fullTopic, ex.getMessage(), ex);
            ScriptShaHelper.reloadScriptSha(ex, this.redissonClient, this.properties);
        }
    }

    @Override
    public void renewOrRetry(final String readableTabKey,
                             final String timeoutTabKey,
                             final String metadataTabKey,
                             final String fullTopic,
                             final Metadata<V> metadata) {
        Map<String, String> cache = SpartaLuaConstant.getSha(SpartaConstant.RENEW_AND_RETRY);
        final List<Object> keys = Arrays.asList(timeoutTabKey, metadataTabKey, readableTabKey);
        String metadataJson = JacksonHelper.serialize(metadata);
        if (this.properties.isEnableMetadataCompress()) {
            metadataJson = ZipStrHelper.compress(metadataJson);
        }

        try {
            RScript script = this.redissonClient.getScript(new LongCodec());
            script.evalSha(RScript.Mode.READ_ONLY, cache.get(SpartaConstant.LUA_SHA_KEY),
                    RScript.ReturnType.VALUE, keys, metadata.getExpire(), fullTopic, metadataJson);
            SpartaWorkerFeature.interruptWorker();
        } catch (Exception ex) {
            log.error("[Sparta] Renew retry failed to execute, topic: {}, exception: {}",
                    fullTopic, ex.getMessage(), ex);
            ScriptShaHelper.reloadScriptSha(ex, this.redissonClient, this.properties);
        }
    }

    @Override
    public void delSegmentReadableTab(final String fullTopic, final String readableTabKey) {
        Map<String, String> cache = SpartaLuaConstant.getSha(SpartaConstant.DEL_SEGMENT_READABLE_TAB);
        List<Object> keys = Collections.singletonList(readableTabKey);

        try {
            RScript script = this.redissonClient.getScript(new StringCodec());
            script.evalSha(RScript.Mode.READ_ONLY, cache.get(SpartaConstant.LUA_SHA_KEY),
                    RScript.ReturnType.VALUE, keys, fullTopic);
        } catch (Exception ex) {
            log.error("[Sparta] Del segment readable tab failed to execute, topic: {}, exception: {}",
                    fullTopic, ex.getMessage(), ex);
            ScriptShaHelper.reloadScriptSha(ex, this.redissonClient, this.properties);
        }
    }

    @Override
    public void delSegmentFinishedConsumption(String fullTopic,
                                              final String metadataField,
                                              final String readableTabKey,
                                              final String metadataTabKey) {
        Map<String, String> cache = SpartaLuaConstant.getSha(SpartaConstant.DEL_SEGMENT_FINISHED_CONSUMPTION);
        List<Object> keys = Arrays.asList(readableTabKey, metadataTabKey);

        try {
            RScript script = this.redissonClient.getScript(new StringCodec());
            script.evalSha(RScript.Mode.READ_ONLY, cache.get(SpartaConstant.LUA_SHA_KEY),
                    RScript.ReturnType.VALUE, keys, fullTopic, metadataField);
        } catch (Exception ex) {
            log.error("[Sparta] Del segment finished consumption failed to execute, topic: {}, exception: {}",
                    fullTopic, ex.getMessage(), ex);
            ScriptShaHelper.reloadScriptSha(ex, this.redissonClient, this.properties);
        }
    }
}
